package com.example.rocketmq.message;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @version 1.0
 **/
@Component
public class ProducerTransaction {


    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer();
        producer.setNamesrvAddr("101.35.153.254:9876;121.5.23.139:9876;1.15.1.55:9876");
        producer.setProducerGroup("groupTransaction");
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("执行本地事务");
                if (StringUtils.equals("tagA", msg.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("tagB", msg.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else {
                    return LocalTransactionState.UNKNOW;
                }

            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("MQ检查消息tag【"+msg.getTags()+"】的本地事务执行结果");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        producer.start();
        String[] tags = {"tagA","tagB","tagC"};
        for (int i = 0; i < 3; i++) {
            Message message = new Message();
            message.setTopic("topicF");
            message.setTags(tags[i]);
            message.setBody(("消息A"+i).getBytes());
            SendResult sendResult = producer.sendMessageInTransaction(message, null);
            System.out.println(sendResult);
            TimeUnit.SECONDS.sleep(1);
        }

//        producer.shutdown();
    }
}
